package com.iqiyi.pps.epg.core.service.stream;

import com.alibaba.fastjson.JSONObject;
import com.iqiyi.pps.epg.api.model.web.stream.TNetMonitorInfo;
import com.iqiyi.pps.epg.core.dao.channel.ChannelConfigDao;
import com.iqiyi.pps.epg.core.dao.log.StreamMonitorLogDao;
import com.iqiyi.pps.epg.core.dao.log.StreamSpanLogDao;
import com.iqiyi.pps.epg.core.dao.stream.LiveStreamBaseDao;
import com.iqiyi.pps.epg.core.model.channel.ChannelBase;
import com.iqiyi.pps.epg.core.model.sginal.VersionController;
import com.iqiyi.pps.epg.core.model.stream.*;
import com.iqiyi.pps.epg.core.rpc.qipu.impl.QipuManager;
import com.iqiyi.pps.epg.core.service.channel.ChannelService;
import com.iqiyi.pps.epg.core.utils.ApiAlertPlatFormUtil;
import com.iqiyi.pps.epg.core.utils.ApplicationContextUtil;
import com.iqiyi.pps.epg.core.utils.LiveStreamManager;
import org.hibernate.Criteria;
import org.hibernate.Session;
import org.hibernate.criterion.Restrictions;
import org.hibernate.type.StringType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.math.BigInteger;
import java.util.ArrayList;
import java.util.List;

/**
 * Created by Administrator on 2014/12/10.
 */
@Service
@Transactional
public class LiveStreamServiceImpl implements LiveStreamService {
    protected Logger logger = LoggerFactory.getLogger(LiveStreamServiceImpl.class);

    @Autowired
    private LiveStreamBaseDao liveStreamBaseDao;

    @Autowired
    private StreamSpanLogDao streamSpanLogDao;

    @Autowired
    private StreamMonitorLogDao streamMonitorLogDao;

    @Autowired
    private LiveChannelStreamService liveChannelStreamService;

    @Autowired
    private ChannelService channelService;

    @Autowired
    private ChannelConfigDao channelConfigDao;



    @Override
    public void save(LiveStream liveStream) {
        liveStreamBaseDao.save(liveStream);
    }

    @Override
    public void edit(LiveStream liveStream) {
        liveStreamBaseDao.merge(liveStream);
    }

    @Override
    public void delete(Long fatherId, String streamName, String formatType) {  //逻辑删除
        liveStreamBaseDao.delete(fatherId, streamName, formatType);
    }

    @Override
    public void delete(Long id) {  //物理删除
        liveStreamBaseDao.delete(id);
    }

    @Override
    public List<LiveStream> get(Long fatherId, String streamName) {
        return liveStreamBaseDao.get(fatherId,streamName);
    }

    @Override
    public LiveStream get(Long fatherId, String streamName, String formatType) {
        return liveStreamBaseDao.get(fatherId,streamName,formatType);
    }

    @Override
    public LiveStream getById (Long id) {
        return liveStreamBaseDao.get(id);
    }
    @Override
    public LiveStream getByDefault(Long fatherId, String formatType, Boolean isDefault) {
        return liveStreamBaseDao.getByDefault(fatherId, formatType, isDefault);
    }

    @Override
    public List<LiveStream> getByFormatType(Long fatherId, String formatType) {
        return liveStreamBaseDao.getByFormatType(fatherId, formatType);
    }

    @Override
    public List<LiveStream> get(Long fatherId) {
        Criteria criteria = liveStreamBaseDao.createCriteria();
        criteria.add(Restrictions.eq("fatherId", fatherId));
        criteria.add(Restrictions.eq("status", LiveStream.STATUS_START));
        return liveStreamBaseDao.find(criteria);
    }

    // 更新老奇谱数据
//    @Override
//    public void writeToTvQipu(Long fatherId, Long oldQipuId, List<LiveChannelStream> liveChannelStreams) {
//        QipuManager qipuManager = (QipuManager) ApplicationContextUtil.getBean(QipuManager.class);
//        if (liveChannelStreams == null) {  //如果为null则清空所有老qipu数据
//            qipuManager.addTvLiveStreamListToQipu(oldQipuId, fatherId, new ArrayList<LiveStream>());
//        } else {
//            List<LiveStream> liveStreams = new ArrayList<LiveStream>();
//            for (LiveChannelStream liveChannelStream : liveChannelStreams) {
//                List<LiveStream> _temp = getByStatus(fatherId, liveChannelStream.getStreamName(), 1);
//                liveStreams.addAll(_temp);
//            }
//            qipuManager.addTvLiveStreamListToQipu(oldQipuId, fatherId, liveStreams);
//        }
//    }

    /**
     * 获取指定状态的列表
     * @param status
     * @return
     */
    @Override
    public List<LiveStream> get(int status) {
        return liveStreamBaseDao.findBy("status", status);
    }

    @Override
    public List<LiveStream> getByStatus(Long fatherId, String streamName, int status) {
        return liveStreamBaseDao.getByStatus(fatherId, streamName, status);
    }


    @Override
    public List<LiveStream> getNeedCheckList() {
        StringBuilder hql = new StringBuilder();
        List<LiveStream> result = new ArrayList<>();


        try {
            hql.append("SELECT {live.*}, {channel_base.*}, {channel.*} FROM live_stream live, live_channel_stream channel ");
            hql.append("LEFT JOIN channel_base on father_id = channel_base.id WHERE ");
            hql.append("live.father_id = channel.father_id AND live.stream_name = channel.stream_name AND ");
            hql.append("(live.format_type = 'FLV' OR live.format_type = 'TS') AND ");
            hql.append("channel.status = 1 AND channel.is_enable = 1 AND live.status = 1 AND channel_base.formal = 1");

            List sqlResult = liveStreamBaseDao.getSession().createSQLQuery(hql.toString())
                    .addEntity("live", LiveStream.class)
                    .addEntity("channel_base", ChannelBase.class)
                    .addEntity("channel", LiveChannelStream.class)
                    .list();

            for (Object single : sqlResult) {
                Object[] obj = (Object[]) single;
                LiveStream liveStream = (LiveStream) obj[0];
                ChannelBase channelBase = (ChannelBase) obj[1];
                LiveChannelStream liveChannelStream = (LiveChannelStream) obj[2];
                liveStream.setChannelLiveType(channelBase.getLiveType());
                liveStream.setStreamType(liveChannelStream.getStreamType());
                result.add(liveStream);
            }
        } catch (Exception e) {
            logger.info("[getNeedCheckList] exception=" + e);
        }

        return result;
    }

    @Override
    public List<StreamCutterIdData> getCutterIdInfo(int maxVal, String sortType, String cutterType) {
        StringBuffer hql = new StringBuffer();
        List<StreamCutterIdData> result = new ArrayList<>();

        hql.append("SELECT c.`key`, c.descr, IFNULL(l.count,0) as 'count' FROM channel_config AS c LEFT JOIN " +
                "(SELECT REPLACE (substr(url, 2, 2), '_', '') AS 'cutter_id', count(*) AS 'count'FROM live_stream WHERE " +
                "format_type = \"" + cutterType + "\" AND `status` != - 1 GROUP BY REPLACE (substr(url, 2, 2), '_', '') ) AS l ON c.`key` = l.cutter_id " +
                "WHERE c.type = 'CutterIdInfo'AND c.`status` = 2 ");
        if(maxVal > 0) {
            hql.append(" AND IFNULL(l.count,0) < " + maxVal);
        }
        hql.append(" ORDER BY l.count "  + sortType + " , c.`key`");

        List sqlResult = liveStreamBaseDao.getSession().createSQLQuery(hql.toString()).list();

        for (Object single : sqlResult) {
            Object[] obj = (Object[]) single;
            StreamCutterIdData temp = new StreamCutterIdData();
            temp.setCutterId(obj[0].toString());
            temp.setDesc((String) obj[1]);
            temp.setCount(obj[2].toString());
            result.add(temp);
        }

        return result;
    }

    @Override
    public void deleteQipu(long liveChannelStreamId, long fatherId, String streamName, LiveStream defaultStream, int streamApiVersion) {
        List<LiveStream> liveStreams = this.get(fatherId, streamName);
        if (liveStreams != null && liveStreams.size() > 0) {
            for (LiveStream liveStream : liveStreams) {
                //删除Qipu数据
                if (liveStream.getStatus() != LiveStream.STATUS_DELETED) {
                    Long qipuId = channelService.getQipuId(liveStream.getFatherId());
                    if (qipuId > 0) {
                        QipuManager qipuManager = (QipuManager) ApplicationContextUtil.getBean(QipuManager.class);
                        qipuManager.delLiveStreamToQipu(qipuId, liveStream.getUuid());
                        if (LiveStream.FORMATTYPE_FLV.equals(liveStream.getFormatType())) {
                            //删除defaultLiveStreamToQipu数据
                            qipuManager.setDefaultLiveStreamToQipu(qipuId, defaultStream);
                        }
                    }
                }
                //删除流数据
                LiveStreamManager.deleteStream(liveStream, liveChannelStreamId, streamApiVersion);
                //逻辑删除库中数据
                this.delete(liveStream.getId());
            }
        }
    }

    @Override
    public void saveStreamSpanLog(StreamSpanLog streamSpanLog) {
        try {
            streamSpanLogDao.save(streamSpanLog);
        } catch (Exception e) {
            logger.info("[saveStreamSpanLog] exception=" + e);
        }
    }

    @Override
    public List<TNetMonitorInfo> getNetMonitorInfo(long channelQipuId){
        StringBuilder sql = new StringBuilder();

        sql.append("select channel_base.name, channel_base.qipu_id, rtmp_server.private_ip , live_channel_stream.stream_type, live_channel_stream.encoded_ip ," +
                "  live_stream.url, cutter_server.private_server_ip , live_stream.format_type " +
                " from channel_base join live_channel_stream on channel_base.id = live_channel_stream.father_id join live_stream" +
                " on live_stream.father_id = channel_base.id  and live_stream.stream_name = live_channel_stream.stream_name" +
                " join cutter_server on substring_index(live_stream.url, \"_\", 1) = cutter_server.cutter_id join rtmp_server " +
                " on live_channel_stream.stream_ip = rtmp_server.public_ip where channel_base.qipu_id = " + channelQipuId +
                " and" +
                " live_channel_stream.`status` > -1" +
                " and live_stream.format_type in('FLV','TS') and live_stream.`status` > -1 order by live_channel_stream.stream_name, stream_type");


        logger.info("[getNetMonitorInfo] sql=" + sql);

        List result = liveStreamBaseDao.getSession().createSQLQuery(sql.toString()).list();


        List<TNetMonitorInfo> ret = new ArrayList<>();
        for (Object single : result) {
            Object[] obj = (Object[]) single;
            logger.info("[getNetMonitorInfo] RtmpIp={}, encodeIp={}", obj[2], obj[4]);
            TNetMonitorInfo temp = new TNetMonitorInfo();
            temp.setChannelName( obj[0].toString());
            temp.setChannelQipuId(((BigInteger) obj[1]).longValue());
            temp.setRtmpIp(obj[2].toString());
            temp.setStreamType(obj[3].toString());
            temp.setEncodeIp(obj[4].toString());
            temp.setCutter(obj[5].toString());
            temp.setCutterIp(obj[6].toString());
            temp.setFormatType(obj[7].toString());

            ret.add(temp);
        }
        return ret;


    }

    @Override
    public void saveStreamMonitorLog(LiveStreamMonitor liveStreamMonitor) {
        try {
            streamMonitorLogDao.save(liveStreamMonitor);
        } catch (Exception e) {
            logger.info("[saveStreamMonitorLog] exception=" + e);
        }

    }

    @Override
    public void emailNotifyUser(String msg) {
        try {
            Session session = channelConfigDao.getSession();

            String hqlUser = "select val from ChannelConfig where type='"+ LiveStreamMonitor.User_TYPE+"' order by key asc";
            List<String> userNames = session.createQuery(hqlUser).list();
            if(userNames.size()>0){
                for(String name:userNames){
                    JSONObject jsonObject = new JSONObject();
                    jsonObject.put("content",msg);
                    //ApiAlertPlatFormUtil.doSend(ApiAlertPlatFormUtil.TOPIC_ID_LIVE_RECORD,ApiAlertPlatFormUtil.SECRET_KEY_EPG,jsonObject.toJSONString(),name,ApiAlertPlatFormUtil.SEND_TYPE_MAIL);
                    ApiAlertPlatFormUtil.doSend(ApiAlertPlatFormUtil.TOPIC_ID_LIVE_RECORD,ApiAlertPlatFormUtil.SECRET_KEY_EPG,jsonObject.toJSONString(),name,ApiAlertPlatFormUtil.SEND_TYPE_CHAT);
                }
            }
        } catch (Exception e) {
            logger.info("[emailNotifyError] exception=" + e);
        }
    }

    @Override
    public String getLiveChannelName(String streamIp, String streamName) {
        return liveChannelStreamService.getByIpAndName(streamIp,streamName);
    }

    @Override
    public List<String> getNotifyErrCode() {
        Session session = channelConfigDao.getSession();
        String hqlCode = "select val from ChannelConfig where type='"+ LiveStreamMonitor.ERROR_TYPE+"' order by key asc";
        List<String> errorCodes = session.createQuery(hqlCode).list();
        return errorCodes;
    }
}
